package com.andox.fn.inf.rpc.rest;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.springframework.web.client.HttpClientErrorException;
import org.springframework.web.client.ResourceAccessException;
import org.springframework.web.client.RestTemplate;

import com.andox.fn.inf.rpc.rest.common.ResponseFuture;


/**   
 * @ClassName:   HttpRemotingService   
 * @Description: TODO  
 * @author:      Andox
 * @date:        2019年8月1日 下午3:51:42   
 *     
 */
public class HttpRemotingService implements RemotingService {

	private RestTemplate restTemplate = new RestTemplate();
	
	private BlockingQueue<Runnable> blockingQueue = new LinkedBlockingQueue<>(2000);
	private ExecutorService executorService = new ThreadPoolExecutor(100, 200, 60, TimeUnit.SECONDS, this.blockingQueue, new ThreadFactory() {
		private AtomicInteger c = new AtomicInteger(0);
		@Override
		public Thread newThread(Runnable r) {
			return new Thread(r, String.format("http-remoting-service-thread-%s", c.incrementAndGet()));
		}
	});
	
	@Override
	public void invokeOneway(RpcRequest request) {
		
		String url = RequestContext.instance().getStringProperty(RequestContextConstants.REQUEST_URL);
		Runnable r = new Runnable() {
			@Override
			public void run() {
				RpcResponse response = null;
				try {
					response = restTemplate.postForObject(url, request, RpcResponse.class);
					if (response == null) {
						response = new RpcResponse(request.getOpaque());
						response.setCode(ResponseCodes.IO_1_ERROR);
						response.setMsg("server response null");
					}else if (ResponseCodes.SUCCESS.equals(response.getCode())){
						// success...
						return;
					}
				} catch (HttpClientErrorException e) {
					// 404 ...
					response = RpcResponse.failed(request.getOpaque(), ResponseCodes.IO_1_ERROR, e.toString());
				}catch (ResourceAccessException e) {
					// 500 ...
					response = RpcResponse.failed(request.getOpaque(), ResponseCodes.IO_2_ERROR, e.toString());
				} catch (Exception e) {
					response = RpcResponse.failed(request.getOpaque(), ResponseCodes.BIZ_ERROR, e.toString());
				}

				response.recreate();
			}
		};

		executorService.execute(r);
	}

	@Override
	public RpcResponse invokeSync(RpcRequest request, int timeoutInMilliseconds) {
		String url = RequestContext.instance().getStringProperty(RequestContextConstants.REQUEST_URL);

		Runnable r = new Runnable() {
			@Override
			public void run() {
				RpcResponse response = null;
				try {
					
					response = restTemplate.postForObject(url, request, RpcResponse.class);
					if (response == null) {
						response = new RpcResponse(request.getOpaque());
						response.setCode(ResponseCodes.IO_1_ERROR);
						response.setMsg("server response null");
					}
				} catch (HttpClientErrorException e) {
					// 404 ...
					response = RpcResponse.failed(request.getOpaque(), ResponseCodes.IO_1_ERROR, e.toString());
				}catch (ResourceAccessException e) {
					// 500 ...
					response = RpcResponse.failed(request.getOpaque(), ResponseCodes.IO_2_ERROR, e.toString());
				} catch (Exception e) {
					response = RpcResponse.failed(request.getOpaque(), ResponseCodes.BIZ_ERROR, e.toString());
				}
				
				ResponseFuture.putResponse(response);
			}
		};

		
		ResponseFuture rf = new ResponseFuture(request);
		rf.setStartTimestampInMilliseconds(System.currentTimeMillis());
		rf.setTimeoutInMilliseconds(timeoutInMilliseconds);
		executorService.execute(r);

		RpcResponse response = rf.get(RequestContext.instance().getIntegerProperty(RequestContextConstants.TIMEOUT_IN_MILLISECONDS));
		
		
		return response;
	}

	@Override
	public void invokeAsync(RpcRequest request, int timeoutInMilliseconds, RemotingCallback callback) {
		String url = RequestContext.instance().getStringProperty(RequestContextConstants.REQUEST_URL);
		RequestContext.instance().putProperty(RequestContextConstants.ASYNC, Boolean.TRUE);
		
		ResponseFuture rf = new ResponseFuture(request);
		rf.setStartTimestampInMilliseconds(System.currentTimeMillis());
		rf.setTimeoutInMilliseconds(timeoutInMilliseconds);
		
		RequestContext.instance().putProperty(RequestContextConstants.RESPONSE_FUTURE, rf);

		Runnable r = new Runnable() {
			@Override
			public void run() {
				RpcResponse response = null;
				try {
					response = restTemplate.postForObject(url, request, RpcResponse.class);
					if (response == null) {
						response = RpcResponse.failed(request.getOpaque(), ResponseCodes.IO_1_ERROR, "server response null");	
					}
				} catch (HttpClientErrorException e) {
					// 404 ...
					response = RpcResponse.failed(request.getOpaque(), ResponseCodes.IO_1_ERROR, e.toString());
				}catch (ResourceAccessException e) {
					// 500 ...
					response = RpcResponse.failed(request.getOpaque(), ResponseCodes.IO_2_ERROR, e.toString());
				} catch (Exception e) {
					response = RpcResponse.failed(request.getOpaque(), ResponseCodes.BIZ_ERROR, e.toString());
				}
				
				ResponseFuture.putResponse(response);
			}
		};

		executorService.execute(r);
	}
}
